-- lua 发送日志到kafka
local cjson=require("cjson")
-- 引用kafka生产者的工具包
local producer = require("resty.kafka.producer")
local broker_list = {
	{ host = "192.168.60.100", port = 9092},
	{ host = "192.168.60.103", port = 9092},
	{ host = "192.168.60.102", port = 9092},
}

-- 定义日志信息
local log_json = {}
log_json["headers"]=ngx.req.get_headers()
log_json["uri_args"] = ngx.req.get_uri_args()  
log_json["body"] = ngx.req.read_body()  
log_json["http_version"] = ngx.req.http_version()  
log_json["method"] =ngx.req.get_method()
log_json["raw_reader"] = ngx.req.raw_header()  
log_json["body_data"] = ngx.req.get_body_data()

-- 序列化成一个字符串
local message=cjson.encode(log_json)
-- 获取生产者id
local productId = ngx.req.get_uri_args()["productId"]
-- 异步发送消息
local async_producer = producer:new(broker_list, {producer_type="async"})
local ok,err = async_producer:send("product_info_log", productId, message)

if not ok then
	ngx.log(ngx.ERR, "kafka send err:", err)
end